package org.apache.ocean.solr.core;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.index.IndexWriter;
import org.apache.lucene.store.Directory;
import org.apache.lucene.store.RAMDirectory;
import org.apache.ocean.solr.core.IndexDocuments.IndexDocument;
import org.apache.solr.schema.IndexSchema;

/**
 * 
 */
public class IndexCreator {
	private Directory directory;
	private long maxSize;
	private int threads; 
	private ExecutorService threadPool;
	private IndexWriter indexWriter;
	private boolean isFinished;
	private List<Future<Object>> futures;
	private IndexSchema schema;

	public IndexCreator(Directory directory, long maxSize, int threads, ExecutorService threadPool) {
		this.directory = directory;
		this.maxSize = maxSize;
		this.threads = threads;
		this.threadPool = threadPool;
		isFinished = false;
	}

	public static class Add {
		private IndexDocument indexDocument;
		private RAMDirectory ramDirectory;
		private IndexSchema indexSchema;

		public Add(RAMDirectory ramDirectory, IndexSchema indexSchema) {
			this.ramDirectory = ramDirectory;
			this.indexSchema = indexSchema;
		}

		public Add(IndexDocument indexDocument, IndexSchema indexSchema) {
			this.indexDocument = indexDocument;
			this.indexSchema = indexSchema;
		}

		public RAMDirectory getRamDirectory() {
			return ramDirectory;
		}

		public IndexDocument getIndexDocument() {
			return indexDocument;
		}

		public IndexSchema getSchema() {
			return indexSchema;
		}
	}

	public void start(Analyzer defaultAnalyzer, BlockingQueue<Add> queue) throws Exception {
		indexWriter = new IndexWriter(directory, false, defaultAnalyzer, true);
		indexWriter.setUseCompoundFile(true);
		indexWriter.setRAMBufferSizeMB(IndexWriter.DISABLE_AUTO_FLUSH);
		indexWriter.setMaxBufferedDocs(IndexWriter.DISABLE_AUTO_FLUSH);
		List<Callable<Object>> callables = new ArrayList<Callable<Object>>(threads);
		for (int x = 0; x < threads; x++) {
			callables.add(new Task(queue));
		}
		futures = threadPool.invokeAll(callables);
	}

	public void create() throws Exception {
		try {
			for (Future<Object> future : futures) {
				if (future.isDone()) {
					try {
						future.get();
					} catch (ExecutionException executionException) {
						Throwable cause = executionException.getCause();
						if (cause instanceof Exception) {
							throw (Exception) cause;
						} else {
							throw new Exception(cause);
						}
					}
				}
				Thread.sleep(10);
			}
			indexWriter.optimize();
		} finally {
			indexWriter.close();
		}
	}
 
	public void setFinished(boolean isFinished) {
		this.isFinished = isFinished;
	}

	private boolean isFinished() {
		if (isFinished)
			return true;
		if (indexWriter.ramSizeInBytes() >= maxSize) {
			isFinished = true;
		}
		return isFinished;
	}
 
	public class Task implements Callable {
		private BlockingQueue<Add> queue;

		public Task(BlockingQueue<Add> queue) {
			this.queue = queue;
		}

		public Object call() throws Exception {
			while (!isFinished()) {
				Add add = queue.poll(3, TimeUnit.MILLISECONDS);
				IndexDocument indexDocument = add.getIndexDocument();
				IndexSchema schema = add.getSchema();
				Analyzer analyzer = schema.getAnalyzer();
				indexWriter.addDocument(indexDocument.toDocument(), analyzer);
			}
			return null;
		}
	}
}
